
#include "communkit.h"
#include "../activite/webcollector.h"
#include "../core/operator_message.h"
#include "../management/channel.h"

#include <zmq.h>
#include <sys/types.h>
#include <net/if.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <linux/sockios.h>
#include <sys/ioctl.h>
#include <arpa/inet.h>
#include <json/json.h>
#include <string>
#include <list>
#include <boost/lexical_cast.hpp>
#include <boost/date_time.hpp>
#include <boost/thread/thread.hpp>
#include <boost/archive/iterators/base64_from_binary.hpp>  
#include <boost/archive/iterators/binary_from_base64.hpp>  
#include <boost/archive/iterators/transform_width.hpp>  
#include "infoconstructor.h"

#include <QDebug>

CommunKit::CommunKit(ProcessRouter::Channel *channel)
    : QObject(0)
    , m_channel(channel)
    , m_hasActive(false)
{
    WebCollector* collector = WebCollector::instance();

    m_pages.set_pid(collector->currentPid());

    m_status.set_pid(collector->currentPid());
    m_status.set_runningcount(collector->getRunningActiviteCount());
    m_status.set_waitcount(collector->getWaitActiviteCount());
    m_status.set_workedcount(collector->getHasWorkedCount());
    m_status.set_isstatic(collector->isStatic());
    m_status.set_state(ProcessRouter::CoreStatus::Initialization);
 
    activationProtocol();
    m_timerId = startTimer(200);
}

CommunKit::~CommunKit()
{
    if (m_timerId)
        killTimer(m_timerId);
}

std::list<std::string> CommunKit::recvNoWait()
{
    std::list<std::string> messlist;
    int times = 100;

    boost::unique_lock<boost::mutex> lock(m_mutex);
    while(1) {
        if (m_msgs.empty())
            break;

        std::string message = m_msgs.front();
        m_msgs.pop();

        messlist.push_back(message);

        if (times == 0)
            break;
        times --;
    }
 
    return messlist;
}

/*
 * Function call message format is a json string like this.
 * {
 *   "clientid" : "10.0.0.12:pid/tid",
 *   "pageid"  : 64823002,
 *   "funcid"  : 22,
 *   "arg0"    : "1",
 *   "arg1"    : "2",
 *   "arg2"    : "3"
 * }
 *
 * Function call return message format 
 * {
 *   "clientid" : "10.0.0.12:pid/tid",
 *   "nodeid"   : "10.0.0.123:pid/tid",
 *   "pageid"   : 23828320,
 *   "funcid"   : 22,
 *   "retval"   : "22332"
 * }
 */
void CommunKit::sendReturnJson(QString clientid, long pageid, long funcid, QString retval, bool encode)
{
    std::string message;
    Json::FastWriter writer;
    Json::Value root(Json::objectValue);

    std::string returnUtf8Str = retval.toUtf8().constData();
    std::cout << __func__ << returnUtf8Str << std::endl;
    std::string encodestr;
    if (encode)
        Base64Encode(returnUtf8Str, &encodestr);
    else
        encodestr = returnUtf8Str;

    root["funcid"] = Json::Value((int)funcid);
    root["retval"] = Json::Value(encodestr);

    message = writer.write(root);

    std::string clientID = clientid.toLatin1().constData();

    ProcessRouter::RPCAnswer *answer = new ProcessRouter::RPCAnswer();
    answer->set_clientid(clientID);
    answer->set_retval(message);
    answer->set_exception(false);

    ProcessRouter::MessagePackage package;
    std::string type_name = ProcessRouter::RPCAnswer::descriptor()->full_name();

    package.set_name(type_name);
    package.set_allocated_answer(answer);

    std::string statemess;
    package.SerializeToString(&statemess);

    ProcessRouter::CMessage mess(statemess);
    //std::cout << __func__ << " | " << getpid() << " | " << message << std::endl;
    m_channel->send(mess);
}

void CommunKit::sendDestroyMessage()
{
    m_status.set_state(ProcessRouter::CoreStatus::Dying);
    m_status.set_vaild(false);
    sendStateInfo();
    if (m_timerId) {
        killTimer(m_timerId);
        m_timerId = 0;
    }
}

void CommunKit::timerEvent(QTimerEvent *event)
{
    uploadState();
}

void CommunKit::uploadState()
{
    WebCollector* collector = WebCollector::instance();

    m_status.set_runningcount(collector->getRunningActiviteCount());
    m_status.set_waitcount(collector->getWaitActiviteCount());
    m_status.set_workedcount(collector->getHasWorkedCount());

    sendStateInfo();
    //sendPageInfo();
}

void CommunKit::updateProxy()
{
    ProcessRouter::IPCCommandProxy *command = new ProcessRouter::IPCCommandProxy();
    command->set_pid(WebCollector::instance()->currentPid());
    command->set_host("");
    command->set_port(0);

    ProcessRouter::MessagePackage package;
    std::string type_name = ProcessRouter::IPCCommandProxy::descriptor()->full_name();
    package.set_name(type_name);
    package.set_allocated_commandproxy(command);

    std::string statemess;
    package.SerializeToString(&statemess);

    ProcessRouter::CMessage mess(statemess);
    m_channel->send(mess);
}

void CommunKit::activationProtocol()
{
    if (m_hasActive)
        return ;

    m_hasActive = true;

    m_status.set_state(ProcessRouter::CoreStatus::Activited);
    m_status.set_vaild(true);
    sendStateInfo();
}

void CommunKit::putRequestQueue(std::string msg)
{
    boost::unique_lock<boost::mutex> lock(m_mutex);
    m_msgs.push(msg);
}

void CommunKit::sendStateInfo()
{
    ProcessRouter::MessagePackage package;
    std::string type_name = ProcessRouter::CoreStatus::descriptor()->full_name();


    package.set_name(type_name);

    ProcessRouter::CoreStatus *status = new ProcessRouter::CoreStatus(m_status);
    package.set_allocated_status(status);

    std::string statemess;
    package.SerializeToString(&statemess);

    ProcessRouter::CMessage mess(statemess);
    m_channel->send(mess);
}

void CommunKit::sendPageInfo()
{
    WebCollector* collector = WebCollector::instance();
    collector->getPageInfos(&m_pages);

    ProcessRouter::MessagePackage package;
    std::string type_name = ProcessRouter::CorePages::descriptor()->full_name();


    package.set_name(type_name);

    ProcessRouter::CorePages *pages = new ProcessRouter::CorePages(m_pages);
    package.set_allocated_pages(pages);

    std::string statemess;
    package.SerializeToString(&statemess);

    ProcessRouter::CMessage mess(statemess);
    m_channel->send(mess);
}

QString CommunKit::protocol()
{
    return m_protocol.c_str();
}

// Return start;wait;memory
std::string CommunKit::getStateInfo()
{
    std::string info;
    WebCollector* collector = WebCollector::instance();

    int running = collector->getRunningActiviteCount();
    int wait = collector->getWaitActiviteCount();
    int freeram = collector->getSystemFreeRamPercent();

    std::string startstr = boost::lexical_cast<std::string>(running);
    std::string waitstr = boost::lexical_cast<std::string>(wait);
    std::string freeramstr = boost::lexical_cast<std::string>(freeram);
    info = startstr + ";" + waitstr + ";" + freeramstr;
    
    return info;
}

bool CommunKit::Base64Encode(const std::string& input, std::string* output)
{ 
    typedef boost::archive::iterators::base64_from_binary<boost::archive::iterators::transform_width<std::string::const_iterator, 6, 8> > Base64EncodeIterator;  
    std::stringstream result;  
    copy(Base64EncodeIterator(input.begin()) , Base64EncodeIterator(input.end()), std::ostream_iterator<char>(result));  
    size_t equal_count = (3 - input.length() % 3) % 3;  
    for (size_t i = 0; i < equal_count; i++) {  
        result.put('=');  
    }  
    *output = result.str();  
    return output->empty() == false;  
}

bool CommunKit::Base64Decode(const std::string& input, std::string* output)
{ 
    typedef boost::archive::iterators::transform_width<boost::archive::iterators::binary_from_base64<std::string::const_iterator>, 8, 6> Base64DecodeIterator; 
    std::stringstream result; 
    try { 
        copy(Base64DecodeIterator(input.begin()) , Base64DecodeIterator(input.end()), std::ostream_iterator<char>(result)); 
    } catch(...) { 
        return false; 
    } 
    *output = result.str().c_str();
    return output->empty() == false;
}

void CommunKit::destroy()
{
    m_status.set_state(ProcessRouter::CoreStatus::Dying);
    sendStateInfo();
}

bool CommunKit::getAvaliablePort(unsigned short &port)
{
    SOCKET sock = socket(AF_INET, SOCK_STREAM, 0);

    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = 0;        // 若port指定为0,则调用bind时，系统会为其指定一个可用的端口号

    int ret = bind(sock, (struct sockaddr*)&addr, sizeof(addr));

    if (0 != ret) {
        closesocket(sock);
        return false;
    }

    struct sockaddr_in connAddr;
    socklen_t len = sizeof(connAddr);
    ret = getsockname(sock, (struct sockaddr*)&connAddr, &len);

    if (0 != ret) {
        closesocket(sock);
        return false;
    }

    port = ntohs(connAddr.sin_port); // 获取端口号

    closesocket(sock);

    return true;
}

std::string CommunKit::listToJson(QList<QString> &list)
{
    std::string json = "[";
    for (int i = 0; i < list.size(); i ++) {
        std::string url = "\"";
        url.append(list.at(i).toLatin1().constData());
        url.append("\"");
        json.append(url);
        if (i != list.size() - 1)
            json.append(",");
    }
    json.append("]");
    return json;
}

